package bigdata.scala.spark.streaming.ycfx.userstatusqueue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util

import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.{HBaseClient, KafkaProducerClient}
import bigdata.scala.spark.streaming.ycfx.model.{EleCurveModel, ElectricMpdel, MyConstants}
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.api.java.JavaInputDStream
import org.apache.spark.streaming.dstream.DStream

/**
 * @author wuzhijiang
 *  获取kafka中电流数据写入hbase中
 */
object ElectricStatusQueue {
  def main(args: Array[String]): Unit = {
    //从kafka获取用户电流数据
    val simpleSparkKafka = new SimpleSparkKafka(args)
    simpleSparkKafka.execute(new SKProcessor {
      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        val baseStream: DStream[String] = stream.map(cr=>cr.value())
        dataShareQueue(baseStream)
      }
    }).start()
  }

  //数据转换 将从kafka获取到的json字符串转为string类型
  def dataShareQueue(baseStream: DStream[String]): Unit ={
    baseStream.foreachRDD(rdd=>{
      rdd.map(kafka => JSON.parseObject(kafka))
        .map(json => {
          val meterId = json.getString("METER_ID")
          val meterAsseton = json.getString("METER_ASSETON")
          val dataDate = json.getString("DATA_DATE")
          val phase = json.getString("PHASE")
          val dotdensity = json.getString("DOTDENSITY")
          val i1  = json.getString("I1")
          val i5  = json.getString("I5")
          val i9  = json.getString("I9")
          val i13 = json.getString("I13")
          val i17 = json.getString("I17")
          val i21 = json.getString("I21")
          val i25 = json.getString("I25")
          val i29 = json.getString("I29")
          val i33 = json.getString("I33")
          val i37 = json.getString("I37")
          val i41 = json.getString("I41")
          val i45 = json.getString("I45")
          val i49 = json.getString("I49")
          val i53 = json.getString("I53")
          val i57 = json.getString("I57")
          val i61 = json.getString("I61")
          val i65 = json.getString("I65")
          val i69 = json.getString("I69")
          val i73 = json.getString("I73")
          val i77 = json.getString("I77")
          val i81 = json.getString("I81")
          val i85 = json.getString("I85")
          val i89 = json.getString("I89")
          val i93 = json.getString("I93")
          val electricMpdel = ElectricMpdel(meterId,meterAsseton,dataDate,phase,dotdensity,i1,i5,i9,i13,i17,i21,i25,i29,i33,i37,i41,i45,i49,i53,i57,i61,i65,i69,i73,i77,i81,i85,i89,i93)
          putUserelectricHbase(electricMpdel)
        })
    })
  }

  // 将用户电流写入hbase
  def putUserelectricHbase(electricMpdel: ElectricMpdel) = {
      val dataDate:String = LocalDateTime.parse(electricMpdel.dataDate, DateTimeFormatter.ofPattern("yyyyMMdd")).toString // 获取时间
      putUserelectricHbaseBy24Point(dataDate, electricMpdel)
  }

  // 将24个点用户电流写入hbase
  def putUserelectricHbaseBy24Point(dataDate : String, t : ElectricMpdel): Unit = {
    putUserElectricHbaseByPoint(dataDate, t.phase, 5, t.i5, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 9, t.i9, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 13, t.i13, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 17, t.i17, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 21, t.i21, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 25, t.i25, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 29, t.i29, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 33, t.i33, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 37, t.i37, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 41, t.i41, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 45, t.i45, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 49, t.i49, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 53, t.i53, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 57, t.i57, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 61, t.i61, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 65, t.i65, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 69, t.i69, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 73, t.i73, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 77, t.i77, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 81, t.i81, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 85, t.i85, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 89, t.i89, t.meterId, t.meterAsseton)
    putUserElectricHbaseByPoint(dataDate, t.phase, 93, t.i93, t.meterId, t.meterAsseton)
  }

  // 将一个点的数据写入hbase
  def putUserElectricHbaseByPoint(dataDate : String, phase : String, point : Int, electricValue : String, meterId : String, meterAssetno : String): Unit = {
    //判断数据是否为空
    if(!(electricValue==null||electricValue.equals(""))) {
      // 通过点数生成rowkey放入实体类中
      val eleCurveModel = getRowkey(dataDate, point)
      //列族
      val columnFamily = ""
      // 从hbase获取用户电流
      var eleCurve = getUserElectricHbase(eleCurveModel.rowkey)
      // 生成用户电流数据
      val model: EleCurveModel = genUserElectric(eleCurve, phase, point, electricValue, meterId, meterAssetno)
      // 将正常的用户电流写入hbase
      putElectricPowerHbase(MyConstants.tab_user_electric,eleCurveModel.rowkey,model)
      //判断电流是否异常  bool=false是异常
      val bool = getAbnormalElePower(model)
      if (bool==false){
        //说明电流异常  将异常电流数据写入hbase异常电流表
        putAbnoElePowerHbase(MyConstants.tab_user_electric_abnormal,eleCurveModel.rowkey,model)
        //将异常电流数据下发到kafka
        putAbnoElePowerKafka(model)
      }

    }
  }

  //将正常电流数据写入hbase
  def putElectricPowerHbase(tableName:String,rowKey:String,eleCurveModel: EleCurveModel): Unit ={
    //将实体类转换为json字符串
    val eleCurveModelStr: String = JSON.toJSONString(eleCurveModel)
    /*val elePowerMap:util.HashMap[String,String] = new util.HashMap[String,String]()
    elePowerMap.put("meterId",model.meterId)
    elePowerMap.put("meterAssetno",model.meterAssetno)
    elePowerMap.put("aphase",model.aphase)
    elePowerMap.put("bphase",model.bphase)
    elePowerMap.put("cphase",model.cphase)
    elePowerMap.put("zphase",model.zphase)*/
    HBaseClient.getInstance().putCfDefault(tableName,rowKey,eleCurveModelStr)
  }

  //判断电流是否异常 零火差大于0.1是异常
  def getAbnormalElePower(eleCurveModel: EleCurveModel): Boolean ={
    var boolean:Boolean = true

    val aphase = eleCurveModel.aphase.toDouble
    val bphase = eleCurveModel.bphase.toDouble
    val cphase = eleCurveModel.cphase.toDouble
    val zphase = eleCurveModel.zphase.toDouble

    var az =0.0
    var bz =0.0
    var cz =0.0
    if(!((aphase==null||aphase.equals("")) && (zphase==null||zphase.equals("")))){
      //取其差值的绝对值
      az = math.abs((zphase-aphase).formatted("%.2f").toDouble)
      if(az > 0.1){
        boolean = false
      }
    }
    if(!((bphase==null||bphase.equals("")) && (zphase==null||zphase.equals("")))){
      //取其差值的绝对值
      bz = math.abs((zphase-bphase).formatted("%.2f").toDouble)
      if(bz > 0.1){
        boolean = false
      }
    }
    if(!((cphase==null||cphase.equals("")) && (zphase==null||zphase.equals("")))){
      //取其差值的绝对值
      cz = math.abs((zphase-cphase).formatted("%.2f").toDouble)
      if(cz > 0.1){
        boolean = false
      }
    }
    boolean
  }

  //将异常电流数据写入hbase
  def putAbnoElePowerHbase(tableName:String,rowKey:String,eleCurveModel: EleCurveModel): Unit ={
    //将实体类转换为json字符串
    val eleCurveModelStr: String = JSON.toJSONString(eleCurveModel)
    HBaseClient.getInstance().putCfDefault(tableName,rowKey,eleCurveModelStr)
  }

  //将异常电流数据下发通知至kafka
  def putAbnoElePowerKafka(model: EleCurveModel): Unit ={
    val abnoEleMap:util.Map[Object,Object] = new util.HashMap[Object, Object]
    abnoEleMap.put("meterId", model.meterId)
    abnoEleMap.put("meterAssetno", model.meterAssetno)
    abnoEleMap.put("aphase", model.aphase)
    abnoEleMap.put("bphase", model.bphase)
    abnoEleMap.put("cphase", model.cphase)
    KafkaProducerClient.getInstance().sendMessage(MyConstants.tpc_ele_power_abnormal,abnoEleMap)
  }

  // 生成rowkey,point代表是第几个点 i1是第一个点 以此类推
  def getRowkey(dataDate : String, point : Int) = {
    var eleCurveModel:EleCurveModel = null
    val hourNumber = (point - 1) * 15 / 60; //十五分钟一个点，求出来的就是第几个小时
    val dataDateTime = dataDate + (if (hourNumber<10) "0"+hourNumber.toString else hourNumber.toString)
    eleCurveModel.rowkey=dataDateTime
    eleCurveModel
  }

  // 根据rowkey获取hbase中用户电流
  def getUserElectricHbase(rowkey : String):EleCurveModel = {
    var eleCurveModel:EleCurveModel = null
    //如果hbase有则取hbase中的数据
    if (HBaseClient.getInstance().exists(MyConstants.tab_user_electric, rowkey)) {
      var eleCurve = HBaseClient.getInstance().getCfDefault(MyConstants.tab_user_electric,rowkey)
      //将取出的json数据转为到EleCurveModel类中
      eleCurveModel = JSON.parseObject(eleCurve,classOf[EleCurveModel])
    }
    eleCurveModel
  }

  // 生成电流数据
  def genUserElectric(eleCurveModel :EleCurveModel , phase : String, point : Int, electricValue : String, meterId : String, meterAssetno : String): EleCurveModel = {
    // 如果不存在用户电流则初始化用户电流
    if (eleCurveModel == null) {
      var eleCurveModel:EleCurveModel = null
      eleCurveModel.meterId = meterId
      eleCurveModel.meterAssetno = meterAssetno
    }
    phase match {
      case "1" => {
        eleCurveModel.aphase = electricValue
      }
      case "2" => {
        eleCurveModel.bphase = electricValue
      }
      case "3" => {
        eleCurveModel.cphase = electricValue
      }
      case "0" => {
        eleCurveModel.zphase = electricValue
      }
    }
    eleCurveModel
  }
}
